package ds_industry_2025.ds.ds02.T4

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{OneHotEncoder, StandardScaler, VectorAssembler}
import org.apache.spark.ml.linalg.SparseVector
import org.apache.spark.ml.linalg.DenseVector
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{asc, dense_rank, expr}

import java.util.Properties

/*
    2、根据Hudi的dwd库中相关表或MySQL中shtd_store中相关商品表（sku_info），获取id、spu_id、price、weight、tm_id、
    category3_id 这六个字段并进行数据预处理，对price、weight进行规范化(StandardScaler)处理，对spu_id、tm_id、
    category3_id进行one-hot编码处理（若该商品属于该品牌则置为1，否则置为0）,并按照id进行升序排序，在集群中输出第一条数据
    前10列（无需展示字段名），将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
字段	类型	中文含义	备注
id	double	主键
price	double	价格
weight	double	重量
spu_id#1	double	spu_id 1	若属于该spu_id，则内容为1否则为0
spu_id#2	double	spu_id 2	若属于该spu_id，则内容为1否则为0
.....	double
tm_id#1	double	品牌1	若属于该品牌，则内容为1否则为0
tm_id#2	double	品牌2	若属于该品牌，则内容为1否则为0
……	double
category3_id#1	double	分类级别3 1	若属于该分类级别3，则内容为1否则为0
category3_id#2	double	分类级别3 2	若属于该分类级别3，则内容为1否则为0
……

结果格式如下：
--------------------第一条数据前10列结果展示为：---------------------
1.0,0.892346,1.72568,0.0,0.0,0.0,0.0,1.0,0.0,0.0
 */
object t2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t2")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    //  todo 隐式转换
    import spark.implicits._

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    val sku_info=spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","sku_info",conn)
      .select("id","spu_id","price","weight","tm_id","category3_id")




    //  todo 首先将需要one——hot编码的字段进行索引化

    val source = sku_info
      .withColumn(
        "spu_id_index",
        dense_rank().over(Window.orderBy("spu_id")) - 1
      )
      .withColumn(
        "tm_id_index",
        dense_rank().over(Window.orderBy("tm_id")) - 1
      )
      .withColumn(
        "category3_id_index",
        dense_rank().over(Window.orderBy("category3_id"))-1
      )

    //  todo 将这张表保存，给下一题的推荐系统使用
    source.write.format("hive").mode("overwrite")
      .saveAsTable("tzgc.t1")

    //  todo 设置one-hot编码转换器
    val one_hot_encoder = new OneHotEncoder()
      .setInputCols(Array("spu_id_index","tm_id_index","category3_id_index"))
      .setOutputCols(Array("spu_id_o","tm_id_o","category3_id_o"))
      .setDropLast(true)

    //  todo 将需要标准化处理的字段转化为特征向量,为了标准化做处理
    //  todo 需要注意的是这里为了方便做题，需要把字段一个一个转化为特征向量，然后一个一个标准化
    val vector_assembler1 = new VectorAssembler()
      .setInputCols(Array("price"))
      .setOutputCol("price_v")

    val vector_assembler2=new VectorAssembler()
      .setInputCols(Array("weight"))
      .setOutputCol("weight_v")

    //  todo 设置标准化器，对特征向量进行标准化
    val scaler1 = new StandardScaler()
      .setInputCol("price_v")
      .setOutputCol("price_s")
      .setWithMean(true)

    val scaler2=new StandardScaler()
      .setInputCol("weight_v")
      .setOutputCol("weight_s")
      .setWithMean(true)

    //  todo 设置机器学习流水线
    val pipeline = new Pipeline()
      .setStages(Array(vector_assembler1,vector_assembler2,scaler1,scaler2,one_hot_encoder))
      .fit(source)

    //  todo 定义对密集向量和稀疏向量的处理方法
    //  稀疏向量
    spark.udf.register(
      "vector_to_array",
      (r:SparseVector) => r.toArray.mkString(",")
    )

    //  密集向量
    spark.udf.register(
      "vector",
      (r:DenseVector) => r.apply(0)
    )

    //  todo 使用流水线转换器转换数据，并且应用向量处理函数得到最后的结果
    val result = pipeline.transform(source)
      .withColumn("price_s", expr("vector(price_s)"))
      .withColumn("weight_s", expr("vector(weight_s)"))
      .withColumn("spu_id_o", expr("vector_to_array(spu_id_o)"))
      .withColumn("tm_id_o", expr("vector_to_array(tm_id_o)"))
      .withColumn("category3_id_o", expr("vector_to_array(category3_id_o)"))
      .select("id", "price_s", "weight_s", "spu_id_o", "tm_id_o", "category3_id")
      .orderBy(asc("id"))
      .limit(1)

    // todo result.collect得到是一个Array[Row]类型的数据，所以这里用foreach
    //  todo split是分割字符串，不能按照单字符分割,所以flatMap(r => r.toString.split(","))就会把这一行的数据转换成
    //  todo 一个Array[String]的类型，然后就可以使用take(10)
    println("--------------------第一条数据前10列结果展示为：--------------------")
    result.collect().foreach(
      r => {
        println(r.toSeq.flatMap(r => r.toString.split(",")).take(10).mkString(","))
      }
    )


    spark.close()
  }

}
